package com.leilei;

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author lei
 * @version 1.0
 * @date 2021/3/20 18:06
 * @des flink table 操作表一样操作数据
 */
public class Flink_SQL_1_demo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
        DataStreamSource<GirlFriend> scFriendList = env.fromCollection(Arrays.asList(new GirlFriend("小丽", "A", 22),
                new GirlFriend("小丽于", "C", 24),
                new GirlFriend("小胡", "D", 23)));
        //DataStream 转临时视图
        tableEnv.createTemporaryView("sc", scFriendList,$("name"),$("cup"),$("age"));
        Table table = tableEnv.sqlQuery("select * from sc where age>=23 or name like '%丽%'");
        DataStream<GirlFriend> girlFriendDataStream = tableEnv.toAppendStream(table, TypeInformation.of(GirlFriend.class));
        girlFriendDataStream.print();
        //方式二...........
        //直接将dataStream转为 table
        Table scTable = tableEnv.fromDataStream(scFriendList);
        Table resultTable = tableEnv.sqlQuery("select * from "+scTable+" where  name like '%丽%' and cup='C'");
        DataStream<GirlFriend> result = tableEnv.toAppendStream(resultTable, TypeInformation.of(GirlFriend.class));
        result.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }


    }
}
